package study.bigdata.flink.state;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.FlinkException;

import java.util.concurrent.TimeUnit;

@Slf4j
public class CkpointJob {

    @SneakyThrows
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // check point 配置
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig config = env.getCheckpointConfig();
        config.setMinPauseBetweenCheckpoints(1000);
        config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 状态后端
        StateBackend fsStateBackend = new FsStateBackend("file:///d:/ckTest");
        env.setStateBackend(fsStateBackend);
        env.disableOperatorChaining();

        //
        env.addSource(new CkpointSourceFunction())
                .addSink(new CkpointSinkFunction())
                .uid("ssssssssss");
        env.execute("Flink-Checkpoint-Job");
    }

    @Slf4j
    public static class CkpointSinkFunction extends RichSinkFunction<String> {
        @Override
        public void invoke(String value, Context context) throws Exception {
            log.info(">>>>>> invoke");
            TimeUnit.SECONDS.sleep(10);
            int i = RandomUtils.nextInt(0, 10);
            if (i / 2 == 0) {
                log.info("fffffffffffffffffffffffffffffffff");
                throw new FlinkException("throw exception and restart");
            }
            log.info("[sink] success value {}", value);
        }
    }
}
